package com.etlmaster.executor.steps.datax;

import java.util.HashMap;
import java.util.Map;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.JobPool;
import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.utils.DataCollectionUtils;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("dbwriter")
public class DbWriterStep  extends Step{
	private DbWriterStepMeta meta;
	public DbWriterStep(StepMeta meta) {
		super(meta);
		this.meta=(DbWriterStepMeta) meta;
	}

	@Override
	protected int handle() {
		
		DataSourceDefine def = DataSourceHolder.getInstance()
				.getDataSourceDefine(meta.getDsName());
		String dsType=DataCollectionUtils.getDsType(
				meta.getDsName());
		Map<String, String> data = new HashMap<String, String>();
		data.put("url", def.getUrl());
		data.put("username",def.getUsername());
		data.put("password",def.getPassword());
		data.put("table", meta.getTargetTable());
		data.put("columns", meta.getColumns());
		data.put("concurrency", meta.getConcurrency());

		Map<String,String> rtnMap= DataxWriter.runDatax(this.hashCode(), dsType+"writer", JobPool.getVariable(meta.getSession_id()).get("readerjobconf").toString(), data);
		meta.getStepState().getReturnValue().put("logInfo", rtnMap.get("jobInfo"));
		return Integer.valueOf(rtnMap.get("status"));
		
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}


}
